<?php
declare (strict_types=1);

namespace app\command;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use think\console\Command;
use think\console\Input;
use think\console\Output;

class RpcServer extends Command
{
    protected function configure()
    {
        // 指令配置
        $this->setName('rpcserver')
            ->setDescription('the rpcserver command');
    }

    protected function execute(Input $input, Output $output)
    {
        // 队列名
        $queue = 'rpc_queue';
        // 建立连接
        $connection = new AMQPStreamConnection('127.0.0.1', '5672', 'guest', 'guest', '/');
        // 获取信道
        $channel = $connection->channel();
        // 声明创建队列
        $channel->queue_declare($queue, false, false, false, false);

        // 公平调度
        $channel->basic_qos(null, 1, null);

        // 消息消费
        $channel->basic_consume($queue, '', false, false, false, false, function ($msg) use ($output) {
            // 接收到RPC客户端发送的信息
            $output->writeln('received:' . $msg->body . PHP_EOL);

            // 执行方法，回调队列
            $reply = new AMQPMessage('rpc server reply message', [
                'correlation_id' => $msg->get('correlation_id')
            ]);

            $msg->delivery_info['channel']->basic_publish(
                $reply, '', $msg->get('reply_to'));

            // 确认消息处理完成
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        });

        while (count($channel->callbacks)) {
            $channel->wait();
        }

        // 关闭信道
        $channel->close();
        // 关闭连接
        $connection->close();
    }
}
